package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.UserLoginBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

public class DwsUserUserLoginWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsUserUserLoginWindow().start(Constant.TOPIC_DWD_TRAFFIC_PAGE,Constant.DWS_USER_USER_LOGIN_WINDOW,4,10024);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        // 1. 先过滤出所有登录记录

        //elt
        SingleOutputStreamOperator<JSONObject> eltStream = eltStream(dataStreamSource);

//        eltStream.print();
        // 2. 分组 独立和回流

        SingleOutputStreamOperator<UserLoginBean> userLoginBeanSingleOutputStreamOperator = uv_hlStream(eltStream);


        // 3.水位线开窗聚合

        //水位线
        SingleOutputStreamOperator<UserLoginBean> waterStream = WaterStream(userLoginBeanSingleOutputStreamOperator);

        SingleOutputStreamOperator<UserLoginBean> streamOperator = reduce_winow(waterStream);
        streamOperator.print();

        // 4.写入doris
        streamOperator.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSinks(Constant.DWS_USER_USER_LOGIN_WINDOW));



    }

    private static SingleOutputStreamOperator<UserLoginBean> reduce_winow(SingleOutputStreamOperator<UserLoginBean> waterStream) {
        return waterStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<UserLoginBean>() {
                    @Override
                    public UserLoginBean reduce(UserLoginBean t1, UserLoginBean t2) throws Exception {
                        t1.setUuCt(t1.getUuCt() + t2.getUuCt());
                        t1.setBackCt(t1.getBackCt() + t2.getBackCt());

                        return t1;
                    }
                }, new AllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<UserLoginBean> iterable, Collector<UserLoginBean> collector) throws Exception {

                        long start = timeWindow.getStart();

                        long end = timeWindow.getEnd();

                        Iterator<UserLoginBean> iterator = iterable.iterator();
                        UserLoginBean userLoginBean = iterator.next();
                        userLoginBean.setStt(DateFormatUtil.tsToDateTime(start));
                        userLoginBean.setEdt(DateFormatUtil.tsToDateTime(end));

                        userLoginBean.setCurDate(DateFormatUtil.tsToDateTime(System.currentTimeMillis()));
                        collector.collect(userLoginBean);
                    }
                });
    }

    private static SingleOutputStreamOperator<UserLoginBean> WaterStream(SingleOutputStreamOperator<UserLoginBean> userLoginBeanSingleOutputStreamOperator) {
         return userLoginBeanSingleOutputStreamOperator.assignTimestampsAndWatermarks(WatermarkStrategy
                .<UserLoginBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getTs()));
    }

    private static SingleOutputStreamOperator<UserLoginBean> uv_hlStream(SingleOutputStreamOperator<JSONObject> eltStream) {
         return eltStream.keyBy(x -> x.getJSONObject("common").getString("uid"))
                .process(new ProcessFunction<JSONObject, UserLoginBean>() {

                    ValueState<String> state;

                    @Override
                    public void open(Configuration parameters) {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("state", String.class);
                        state = getRuntimeContext().getState(stateDescriptor);

                    }

                    @Override
                    public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, UserLoginBean>.Context context, Collector<UserLoginBean> collector) throws Exception {

                        String value = state.value();
                        JSONObject common = jsonObject.getJSONObject("common");

                        JSONObject page = jsonObject.getJSONObject("page");

                        Long ts = jsonObject.getLong("ts");

                        String tsToDate = DateFormatUtil.tsToDate(ts);


                        //独立和回流

                        long uv = 0L;
                        long hl = 0L;

                        if (value==null || !value.equals(tsToDate)) {
                            uv = 1L;
                            state.update(tsToDate);
                        }

                        //今天登录 日期减去上次登录日期 大于7天 为回流
                        if(value!=null){
                            if (ts - DateFormatUtil.dateToTs(value) > (7 * 24 * 60 * 60 * 1000)) {
                                hl = 1L;
                                state.update(tsToDate);
                            }
                        }
                        if (uv != 0 || hl != 0) {

                            collector.collect(UserLoginBean.builder()
                                    .backCt(hl)
                                    .uuCt(uv)
                                    .ts(ts)
                                    .curDate(tsToDate)
                                    .build());
                        }
                    }
                });
    }

    private static SingleOutputStreamOperator<JSONObject> eltStream(DataStreamSource<String> dataStreamSource) {
         return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);
                    if (jsonObject != null) {

                        try {
                            JSONObject page = jsonObject.getJSONObject("page");

                            JSONObject common = jsonObject.getJSONObject("common");

                            if (page != null && common != null) {
                                Long ts = jsonObject.getLong("ts");

                                String uid = common.getString("uid");

                                if (uid != null && ts != null) {
                                    String lastPageId = page.getString("last_page_id");

                                    //两种登录方式：1.uid不为空，而上一页 为空，表示 已经登录过，自动登录 2. uid部位空，上一页为登录页面，表示 第一次登录成功跳转到 其他页面
                                    if (lastPageId == null || ("login".equals(lastPageId))) {

                                        collector.collect(jsonObject);
                                    }
                                }
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        });
    }
}
